package com.mango.ch13;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.broadcast.Broadcast;
import scala.Tuple2;

import java.util.*;

public class SparkKNNJob {
    private static String s_inputPath = "d:/HadoopData/input/s_KNN.txt";
    private static String r_inputPath = "d:/HadoopData/input/r_KNN.txt";
    private static String outPutPaht = "d:/HadoopData/output";


    static class Ponit {
        private double x;
        private double y;
        private String lable;

        public Ponit(double x, double y, String lable) {
            this.x = x;
            this.y = y;
            this.lable = lable;
        }

        public Ponit() {
        }

        public double getX() {
            return x;
        }

        public void setX(double x) {
            this.x = x;
        }

        public double getY() {
            return y;
        }

        public void setY(double y) {
            this.y = y;
        }

        public String getLable() {
            return lable;
        }

        public void setLable(String lable) {
            this.lable = lable;
        }

        @Override
        public String toString() {
            return "Ponit{" +
                    "x=" + x +
                    ", y=" + y +
                    ", lable='" + lable + '\'' +
                    '}';
        }

        @Override
        public boolean equals(Object obj) {
            if (this == obj) return true;
            if (obj == null || getClass() != obj.getClass()) return false;
            Ponit other = (Ponit) obj;
            double equalX = this.x - other.getX();
            double equalY = this.y - other.getY();
            boolean equalLable = false;
            if (this.lable == null && other.getLable() == null) equalLable = true;
            else if (this.lable != null && other.getLable() != null) {
                equalLable = this.lable.equals(other.getLable());
            } else
                equalLable = false;

            if (equalX == 0 && equalY == 0 && equalLable) return true;
            else return false;

        }
    }

    /**
     * knn 分析
     *
     * @param k         一个大于0的整数，表示要查询几个相邻的对象作为分类参考
     * @param query     想要分类的对象
     * @param trainSets 已经分类的训练数据集
     */
    public static void knn(int k, Ponit query, List<Ponit> trainSets) {
        //创建长度为k的一个固定的有序映射
        //将距离映射到一个训练点
        SortedMap<Double, Ponit> map = new TreeMap<Double, Ponit>();
        for (Ponit train :
                trainSets) {
            //计算测试点到训练点的距离
            double dis = calDistance(query, train);
            //将训练点插入有序的数据集
            //训练点不在查询点的K个紧邻中
            //将其删除
            if (map.size() > k) {
                double lastN = map.lastKey();
                if (dis < lastN)
                    map.put(dis, train);
            } else
                map.put(dis, train);
        }
        //对k个紧邻完成多数投票
        //并为将查询点分类到相应标签集合
        query.setLable(majortyVote(map, k));
    }

    private static String majortyVote(Map<Double, Ponit> map, int k) {
        return null;
    }

    private static double calDistance(Ponit query, Ponit train) {
        double sum = 0;
        sum += (query.getX() - train.getX()) * (query.getX() - train.getX());
        sum += (query.getY() - train.getY()) * (query.getY() - train.getY());
//        sum += (query.getZ() - train.getZ()) * (query.getZ() - train.getZ());
        sum = Math.sqrt(sum);
        return sum;
    }

    public static void main(String[] args) {
        SparkConf sconf = new SparkConf();
        sconf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer");
        sconf.setAppName("knn");
        sconf.setMaster("local");
        JavaSparkContext jsc = new JavaSparkContext(sconf);
        //广播参数  设置紧邻的个数
        final Broadcast<Integer> broadcastK = jsc.broadcast(15);
        //数据集R
        List<Tuple2<String, String>> R = new ArrayList<>();
        R.add(new Tuple2<>("r1", "R1"));
        R.add(new Tuple2<>("r2", "R2"));
        R.add(new Tuple2<>("r3", "R3"));
        //数据集S
        List<Tuple2<String, String>> S = new ArrayList<>();
        S.add(new Tuple2<>("s1", "S1"));
        S.add(new Tuple2<>("s2", "S2"));
        S.add(new Tuple2<>("s3", "S3"));
        S.add(new Tuple2<>("s4", "S4"));
        /** Distribute a local Scala collection to form an RDD. */
        //将一个本地的list集合作为Rdd分发
        JavaPairRDD<String, String> rdd = jsc.parallelizePairs(R);
        JavaPairRDD<String, String> sdd = jsc.parallelizePairs(S);
        //完成R and S  的笛卡尔乘积
        JavaPairRDD<Tuple2<String, String>, Tuple2<String, String>> cart = rdd.cartesian(sdd);
        //创建初始化数据
        JavaRDD<String> r_lines = jsc.textFile(r_inputPath);
        JavaRDD<String> s_lines = jsc.textFile(s_inputPath);
        JavaRDD<Ponit> rrdd = r_lines.map(new Function<String, Ponit>() {
            @Override
            public Ponit call(String v1) throws Exception {
                Ponit ponit = splitOnToListOfDouble(v1, ",");
                return ponit;
            }
        });

        JavaRDD<Ponit> srdd = s_lines.map(new Function<String, Ponit>() {
            @Override
            public Ponit call(String v1) throws Exception {
                Ponit ponit = splitOnToListOfDouble(v1, ",");
                return ponit;
            }
        });
        //创建笛卡尔乘积
        JavaPairRDD<String, Ponit> cartesian = r_lines.cartesian(srdd);
        cartesian.saveAsTextFile(outPutPaht + "/2");
        //笛卡尔乘积之后对其中的值进行规约一下  求neareastK
        JavaPairRDD<String, Iterable<Ponit>> ponitIntegerJavaPairRDD = cartesian.groupByKey();
        ponitIntegerJavaPairRDD.saveAsTextFile(outPutPaht + "/3");
        JavaPairRDD<String, String> ponitStringJavaPairRDD = ponitIntegerJavaPairRDD.mapToPair(new PairFunction<Tuple2<String, Iterable<Ponit>>, String, String>() {
            @Override
            public Tuple2<String, String> call(Tuple2<String, Iterable<Ponit>> stringIterableTuple2) throws Exception {
                Ponit key = splitOnToListOfDouble(stringIterableTuple2._1, ",");
                String lable = null;
                SortedMap<Double, Ponit> nearestK = findNearestK(key, stringIterableTuple2._2, broadcastK.getValue());
                lable = buildClssification(nearestK);
                key.setLable(lable);
                return new Tuple2<String, String>(key.toString(), lable);
            }
        });
        ponitStringJavaPairRDD.saveAsTextFile(outPutPaht + "/result");
        List<Tuple2<String, String>> collect = ponitStringJavaPairRDD.collect();
        for (Tuple2<String, String> t :
                collect) {
            System.out.println(t._1.toString() + "所属的标签是 " + t._2);
        }
        jsc.close();

    }


    /**
     * 将输入的每行字符分割 返回一个list
     *
     * @param str
     * @param delimiter
     * @return
     */
    static Ponit splitOnToListOfDouble(String str, String delimiter) {
        String[] tokens = str.split(delimiter);
        Ponit p = new Ponit();
        if (tokens.length < 3) {

            p.setX(Double.valueOf(tokens[0]));
            p.setY(Double.valueOf(tokens[1]));
        } else {
            p.setX(Double.valueOf(tokens[0]));
            p.setY(Double.valueOf(tokens[1]));
            p.setLable(tokens[2]);
        }
        return p;

    }

    /**
     * 找到 query对象 最近的K的紧邻对象
     *
     * @param query
     * @param trainSets
     * @param k
     * @return
     */
    static SortedMap<Double, Ponit> findNearestK(Ponit query, Iterable<Ponit> trainSets, int k) {
        SortedMap<Double, Ponit> nearestK = new TreeMap<>();
        for (Ponit t :
                trainSets) {
            Double distance = calDistance(query, t);
            nearestK.put(distance, t);
            //只保留K个紧邻
            if (nearestK.size() > k) {
                nearestK.remove(nearestK.lastKey());
            }
        }
        return nearestK;
    }

    static String buildClssification(SortedMap<Double, Ponit> nearsetK) {
        Map<String, Integer> majortity = new HashMap<>();
        //获取各个紧邻点所属的标签分类
        for (Map.Entry<Double, Ponit> set :
                nearsetK.entrySet()) {
            Ponit p = set.getValue();
            String lable = p.getLable();
            if (majortity.get(lable) == null) {
                majortity.put(lable, 1);
            } else {
                int count = majortity.get(lable) + 1;
                majortity.put(lable, count);
            }

        }
        // 取标签投票较多的作为query的标签
        int max = Integer.MIN_VALUE;
        String queryLable = "";
        for (Map.Entry<String, Integer> set :
                majortity.entrySet()) {
            int num = set.getValue();
            if (num > max) {
                max = num;
                queryLable = set.getKey();
            }
        }
        return queryLable;
    }
}
